package com.joelemon.mockinfluxdbdata.mock.service;

import com.joelemon.mockinfluxdbdata.core.config.MockTask;
import com.joelemon.mockinfluxdbdata.mock.wokers.DBInsertWokerService;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.DelayQueue;

/**
 * @Description: 延迟队列消费者
 * @Author: HJY
 * @Date: 2023/11/20
 */
@Slf4j
public class DelayedQueueConsumer implements Runnable {

    private DBInsertWokerService dbInsertWoker;
    private DelayQueue<MockTask> delayQueue = new DelayQueue<>();


    public DelayedQueueConsumer(DBInsertWokerService dbInsertWoker) {
        this.dbInsertWoker = dbInsertWoker;
    }

    public void putDelayJob(MockTask task) {
        delayQueue.put(task);
    }


    @Override
    public void run() {
        for (;;) {
            try {
                MockTask task = delayQueue.take();
                log.debug(System.currentTimeMillis() + "===>延迟结束：" + task);
                dbInsertWoker.handleWork(task);
            } catch (Exception e) {
                log.error("DelayedQueueConsumer InterruptedException {}" ,e);
            }
        }
    }

    public MockTask[] clearQueue(){
        MockTask[] queueJobs = (MockTask[])delayQueue.toArray();
        delayQueue.clear();
        log.info("DelayedQueueConsumer clearQueue");
        return queueJobs;
    }
}
